package com.bangyu.monitor.core.listener;

import com.bangyu.monitor.core.mqtt.ReceiveMsgTask;
import com.bangyu.monitor.core.schedule.ExportData;
import com.bangyu.monitor.core.schedule.StaticDisplaymentSchedule;
import com.bangyu.monitor.modular.system.entity.MqttTopic;
import com.bangyu.monitor.modular.system.service.MqttTopicService;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;

import java.util.List;
import java.util.Map;

/**
 * @Auther: zhangxue
 * @Date: 19/4/13 16:43
 * @Description:
 */
@Configuration
@Slf4j
public class ApplicationStartUp implements ApplicationListener<ContextRefreshedEvent> {

    private static final String TOPIC = "topic";
    private static final String QOS = "qos";

    @Value("${mqtt.host}")
    private String host;
    @Value("${mqtt.clientId}")
    private String clientId;
    @Value("${mqtt.userName}")
    private String userName;
    @Value("${mqtt.password}")
    private String password;
    @Value("${mqtt.timeOut}")
    private int timeOut;
    @Value("${mqtt.receiveData}")
    private boolean receiveData;

    @Autowired
    private MqttTopicService mqttTopicService;

    @Autowired
    private StaticDisplaymentSchedule staticDisplaymentSchedule;


    @Autowired
    private ExportData exportData;

    private MqttClient client = null;

    @Override
    public void onApplicationEvent(
            ContextRefreshedEvent event) {
//        MDCtest.test();
//        payloadRecoverScheduler.task();
//        InfluxOriginalDbUtil.testPringLog();
        log.info("receive data ?"+receiveData);
        if(!receiveData){
            log.info("this app is only view data");
            return;
        }

        try{
            List<MqttTopic> topics = mqttTopicService.listAll();
            Map formatMap = formatTopicsAndQos(topics);
            String[] topicArr = (String[])formatMap.get(TOPIC);
            int[] qosArr = (int[])formatMap.get(QOS);

            log.info("start up mqtt client begin");
            client = new MqttClient(host,clientId);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(userName);
            options.setPassword(password.toCharArray());
            options.setCleanSession(false);
            options.setConnectionTimeout(timeOut);
            options.setKeepAliveInterval(20);
            options.setAutomaticReconnect(true);
            client.setCallback(new MqttCallback() {
                @Override public void connectionLost(Throwable throwable) {
                    log.error("mqtt client get lost from server.{}",throwable);
                    //将错误日志记录到表
                    //reconnect(client);
//                    LogManager.me().executeLog(LogTaskFactory
//                            .exceptionLog(null, throwable));
                }

                @Override public void messageArrived(String topic,
                        MqttMessage message) throws MqttException {
                    int qos = message.getQos();
                    int id = message.getId();
                    byte[] payloadBytes = message.getPayload();
                    String deviceNo = getDevieNo(topic);
                    Integer deviceType = MqttTopic.getTypeByDeviceNo(topics,deviceNo);

                    log.info("message is arrived ,topic is {},id is{},qos is{},bytes length is {}",topic,id,qos,payloadBytes.length);
                    new ReceiveMsgTask(payloadBytes,deviceNo,deviceType).execute();

                }

                @Override public void deliveryComplete(
                        IMqttDeliveryToken iMqttDeliveryToken) {
                    log.info("deliveryComplete:",iMqttDeliveryToken.isComplete());

                }
            });
             client.connect(options);
            client.subscribe(topicArr,qosArr);

            log.info("start up mqtt client success");
        }catch (MqttException me){
            log.error("mqtt client error",me);
//            reconnect(client);

        }
        catch (Exception e){
            log.error("mqtt client start up failed.",e);
        }
    }

    private Map<String,Object> formatTopicsAndQos(List<MqttTopic> topics){
        int length = topics.size();
        Map formatResult = Maps.newHashMap();
        String[] topicArr = new String[length];
        int[] qosArr = new int[length];
        for(int i=0;i<topics.size();i++){
            topicArr[i] = topics.get(i).getTopic();
            qosArr[i] = topics.get(i).getQos();
        }
        formatResult.put(TOPIC,topicArr);
        formatResult.put(QOS,qosArr);
        return formatResult;

    }

    private void reconnect(MqttClient client){
        int i=0;
        boolean success = client.isConnected();
        try{
            while (!success){
                i++;
                client.reconnect();
                success = client.isConnected();
                log.info("重试重新连接mqtt客户端:第{}次尝试,是否成功:{}",i,success);
                if(i==5){
                    log.error("mqtt client reconnect 5 times failed");
                    break;
                }
                if(!success){
                    Thread.sleep(2000l);
                }

            }
        }catch (Exception e){
            log.error("mqtt client recconnect error",e);
        }
    }

    private String getDevieNo(String topic){
        if(StringUtils.isNotEmpty(topic)){
            try{
                return topic.split("/")[4];
            } catch (Exception e){
                log.error("不能从topic解析出设备号,topic {}",topic);
            }
        }
        return null;
    }


}
